package com.atguigu.day09;

import com.atguigu.bean.OrderEvent;
import com.atguigu.bean.TxEvent;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.util.Collector;

public class Flink05_Process_Join {


    public static void main(String[] args) throws Exception {

        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2.读取支付流数据
//        DataStreamSource<String> orderStrDS = env.readTextFile("input/OrderLog.csv");
        DataStreamSource<String> orderStrDS = env.socketTextStream("hadoop102", 8888);

        //3.读取到账流数据
//        DataStreamSource<String> receiptStrDS = env.readTextFile("input/ReceiptLog.csv");
        DataStreamSource<String> receiptStrDS = env.socketTextStream("hadoop102", 9999);

        //4.将2个流转换为JavaBean对象
        SingleOutputStreamOperator<OrderEvent> orderEventDS = orderStrDS.map(line -> {
            String[] fields = line.split(",");
            return new OrderEvent(Long.parseLong(fields[0]),
                    fields[1],
                    fields[2],
                    Long.parseLong(fields[3]));
        }).filter(data -> data.getTxId() != null).assignTimestampsAndWatermarks(WatermarkStrategy.<OrderEvent>forMonotonousTimestamps()
                .withTimestampAssigner(new SerializableTimestampAssigner<OrderEvent>() {
                    @Override
                    public long extractTimestamp(OrderEvent element, long recordTimestamp) {
                        return element.getEventTime() * 1000L;
                    }
                }));
        SingleOutputStreamOperator<TxEvent> receiptDS = receiptStrDS.map(line -> {
            String[] fields = line.split(",");
            return new TxEvent(fields[0],
                    fields[1],
                    Long.parseLong(fields[2]));
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<TxEvent>forMonotonousTimestamps()
                .withTimestampAssigner(new SerializableTimestampAssigner<TxEvent>() {
                    @Override
                    public long extractTimestamp(TxEvent element, long recordTimestamp) {
                        return element.getEventTime() * 1000L;
                    }
                }));

        //5.使用Connect连接2个流
        ConnectedStreams<OrderEvent, TxEvent> connectedStreams = orderEventDS
                .connect(receiptDS)
                .keyBy(OrderEvent::getTxId, TxEvent::getTxId);

        //6.使用ProcessFunction+状态编程+定时器实现JOIN功能
        SingleOutputStreamOperator<Tuple2<OrderEvent, TxEvent>> result = connectedStreams.process(new MyKeyedProcessFuncJoin());

        //7.打印数据
        result.print();

        //8.启动
        env.execute();

    }

    public static class MyKeyedProcessFuncJoin extends KeyedCoProcessFunction<String, OrderEvent, TxEvent, Tuple2<OrderEvent, TxEvent>> {

        //定义状态用于存放两个流的数据
        private ValueState<OrderEvent> orderEventValueState;
        private ValueState<TxEvent> txEventValueState;

        @Override
        public void open(Configuration parameters) throws Exception {
            orderEventValueState = getRuntimeContext().getState(new ValueStateDescriptor<OrderEvent>("order", OrderEvent.class));
            txEventValueState = getRuntimeContext().getState(new ValueStateDescriptor<TxEvent>("tx", TxEvent.class));
        }

        @Override
        public void processElement1(OrderEvent orderEvent, Context ctx, Collector<Tuple2<OrderEvent, TxEvent>> out) throws Exception {

            System.out.println("OrderWM:" + ctx.timerService().currentWatermark());

            //提取到账数据
            TxEvent txEvent = txEventValueState.value();

            //判断到账数据是否到达
            if (txEvent != null) {
                //结合写出
                out.collect(new Tuple2<>(orderEvent, txEvent));
                //清理状态
                txEventValueState.clear();
            } else {

                System.out.println("注册支付定时器！");

                //将orderEvent写入状态
                orderEventValueState.update(orderEvent);

                //注册定时器用于触发删除状态操作
                TimerService timerService = ctx.timerService();
                //提取数据中的时间
                long ts = orderEvent.getEventTime() * 1000L;
                timerService.registerEventTimeTimer(ts + 10000L);
            }

        }

        @Override
        public void processElement2(TxEvent txEvent, Context ctx, Collector<Tuple2<OrderEvent, TxEvent>> out) throws Exception {

            System.out.println("TxWM:" + ctx.timerService().currentWatermark());

            //提取到账数据
            OrderEvent orderEvent = orderEventValueState.value();

            //判断支付数据是否到达
            if (orderEvent != null) {
                //结合写出
                out.collect(new Tuple2<>(orderEvent, txEvent));
                //清理状态
                orderEventValueState.clear();
            } else {

                System.out.println("注册到账定时器！");

                //将txEvent写入状态
                txEventValueState.update(txEvent);

                //注册定时器用于触发删除状态操作
                TimerService timerService = ctx.timerService();
                //提取数据中的时间
                long ts = txEvent.getEventTime() * 1000L;
                timerService.registerEventTimeTimer(ts + 10000L);
            }
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<OrderEvent, TxEvent>> out) throws Exception {

            System.out.println("调用定时器！");

            OrderEvent orderEvent = orderEventValueState.value();
            if (orderEvent != null) {
                System.out.println(orderEvent.getTxId() + " 触发定时器删除状态");
                orderEventValueState.clear();
            }

            TxEvent tx = txEventValueState.value();
            if (tx != null) {
                System.out.println(tx.getTxId() + " 触发定时器删除状态");
                txEventValueState.clear();
            }

        }
    }

}
